package com.bw.dws;

import com.alibaba.fastjson.JSON;
import com.bw.bean.OrderDetailStats;
import com.bw.fun.AsyncIOFunc;
import com.bw.utils.ClickHouseUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;
import java.util.TimeZone;
import java.util.concurrent.TimeUnit;

public class Test3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.enableCheckpointing(3000);

        DataStream<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("dwd-order-detail-yk5","test671"));

        //stream.print("dwd-order-detail>>>>");

        //OrderDetailStats(stt=null, edt=null, sender_province_id=610000, amount=19.0, sender_province_name=null, order_ct=1, create_time=2023-12-20T06:15:14Z, ts=1703052914000)
        SingleOutputStreamOperator<OrderDetailStats> mapDS = stream.map(new MapFunction<String, OrderDetailStats>() {
            @Override
            public OrderDetailStats map(String s) throws Exception {
                OrderDetailStats od = JSON.parseObject(s, OrderDetailStats.class);//
                // 来一条数据，当做下单数据
                od.setOrder_ct(1);
                //"create_time":"2023-07-11T00:24:52Z"
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                // UTC时区
                sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
                long time = sdf.parse(od.getCreate_time()).getTime();
                // 时间戳
                od.setTs(time);
                return od;
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<OrderDetailStats>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner((event, timestamp) -> event.getTs()));


//        mapDS.print();


        // 迟到数据
        OutputTag<OrderDetailStats> lateTag = new OutputTag<OrderDetailStats>("late-data") {};
        SingleOutputStreamOperator<OrderDetailStats> reduceDS = mapDS.keyBy(x -> x.getSender_province_id())
                .window(TumblingEventTimeWindows.of(Time.minutes(1)))
                .sideOutputLateData(lateTag)
                .reduce(new ReduceFunction<OrderDetailStats>() {
                    @Override
                    public OrderDetailStats reduce(OrderDetailStats x, OrderDetailStats y) throws Exception {
                        x.setOrder_ct(x.getOrder_ct() + y.getOrder_ct());
                        x.setAmount(x.getAmount() + y.getAmount());
                        return x;
                    }
                }, new WindowFunction<OrderDetailStats, OrderDetailStats, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<OrderDetailStats> iterable, Collector<OrderDetailStats> collector) throws Exception {
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String stt = sdf.format(timeWindow.getStart());
                        String edt = sdf.format(timeWindow.getEnd());
                        for (OrderDetailStats od : iterable) {
                            od.setStt(stt);
                            od.setEdt(edt);
                            collector.collect(od);
                        }
                    }
                });
        reduceDS.print();
        //异步IO补充省份名称
//        SingleOutputStreamOperator<OrderDetailStats> asyncDS = AsyncDataStream.unorderedWait(reduceDS, new AsyncIOFunc(), 1000, TimeUnit.MILLISECONDS, 100);
//        asyncDS.print();


        // 把迟到数据放到kafka
        DataStream<OrderDetailStats> sideOutput = reduceDS.getSideOutput(lateTag);
        sideOutput.print();
        sideOutput.map(x->JSON.toJSONString(x)).addSink(MyKafkaUtil.getKafkaProducer("tms_late_topic"));


//        asyncDS.addSink(ClickHouseUtil.sink("insert into order_yk5 (stt,edt,sender_province_name,sender_province_id,amount,order_ct,create_time,ts) values(?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
